#include <stdio.h>
#include <unistd.h>
#include <pthread.h>
#include <stdlib.h>
#include <iostream>
#include <queue>
using namespace std;

#define THREAD_NUM 5
typedef void (*Handler)(int);

//定义线程队列成员类型
class QueType
{
    public:
        QueType(int data,Handler handler)
        {
            data_=data;
            handler_=handler;
        }
        QueType()
        {

        }

        void DealData()
        {
            handler_(data_);
        }
    private:
        //数据和处理方式
        int data_;
        Handler handler_;

};

//线程池
//线程安全队列+多个线程
class ThreadPool
{
    public:
        ThreadPool()
        {
            pthread_mutex_init(&que_lock_,NULL);
            pthread_cond_init(&cons_cond_,NULL);
            pthread_cond_init(&proc_cond_,NULL);
            capacity_ = 10;
            thread_num_ = THREAD_NUM;
            is_quit_ =false;
        }
        ~ThreadPool()
        {
            pthread_mutex_destroy(&que_lock_);
            pthread_cond_destroy(&cons_cond_);
            pthread_cond_destroy(&proc_cond_);
        }
        //创建线程
        int Start()
        {
            for(int i = 0; i<THREAD_NUM;i++) 
            {
                int ret = pthread_create(&tid_[i],NULL,ThreadPollStart,(void*)this);
                if(ret < 0)
                {
                    perror("create");
                    thread_num_--;
                    continue;
                }
            }
            return thread_num_;
        }

        static void* ThreadPollStart(void* arg)
        {
            pthread_detach(pthread_self());
            ThreadPool* tp = (ThreadPool*)arg;
            while(1)
            {
                pthread_mutex_lock(&tp->que_lock_);
                while(tp->que_.empty())
                {
                    if(tp->is_quit_)
                    {
                        tp->thread_num_--;
                        pthread_mutex_unlock(&tp->que_lock_);
                        pthread_exit(NULL);
                    }
                    pthread_cond_wait(&tp->cons_cond_,&tp->que_lock_);
                }
                QueType qt;
                tp->Pop(&qt);
                pthread_mutex_unlock(&tp->que_lock_);
                pthread_cond_signal(&tp->proc_cond_);
                qt.DealData();
            }
            return NULL;
        }

        void Push(QueType& qt)
        {
            pthread_mutex_lock(&que_lock_);
            while(que_.size()>=capacity_)
            {
                pthread_cond_wait(&proc_cond_,&que_lock_);
            }
            que_.push(qt);
            pthread_mutex_unlock(&que_lock_);
            pthread_cond_signal(&cons_cond_);
        }

        void Pop(QueType* qt)
        {
            *qt = que_.front();
            que_.pop();
        }
        void exitThreadPool()
        {
            is_quit_ = true;
            while(thread_num_)
            {
                pthread_cond_signal(&cons_cond_);
            }
        }
    private:
        queue<QueType> que_;
        pthread_mutex_t que_lock_;
        pthread_cond_t cons_cond_;
        pthread_cond_t proc_cond_;
        size_t capacity_;
        int thread_num_;
        pthread_t tid_[THREAD_NUM];

        bool is_quit_;
};

void DeadFunc(int data)
{
    printf("data:%d\n",data);
}

int main()
{
    ThreadPool* tp = new ThreadPool();
    if(tp == NULL)
    {
        exit(1);
    }
    if(tp->Start()<=0)
    {
        exit(2);
    }

    for(int i = 0;i<100;i++)
    {
        QueType qt(i,DeadFunc);
        tp->Push(qt);
    }

    tp->exitThreadPool();
    return 0;
}
